package com.issac.springbootredis.message_queue;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.Set;
import java.util.UUID;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class DelayMessageQueue {
    private Jedis jedis;
    private String queueName;

    //消息入队
    public void queue(Object data){
        Message message = new Message();
        message.setId(UUID.randomUUID().toString());
        message.setData(data);

        try {
            //拿到序列化的消息
            String s = new ObjectMapper().writeValueAsString(message);
            System.out.println("msg publish:" + new Date());
            //消息发送 score延迟五秒
            jedis.zadd(queueName,System.currentTimeMillis()+5000,s);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    //消息消费
    public void loop(){
        while (!Thread.interrupted()){
            //读取score 在0到当前时间戳之间的消息


            /*Set<String> zrangeByScore = null;
            try {
                zrangeByScore  = jedis.zrangeByScore(queueName, 0, System.currentTimeMillis(), 0, 1);
            }catch (ClassCastException e){
                System.out.println("==================");
                e.printStackTrace();
            }*/
            Set<String> zrangeByScore  = jedis.zrangeByScore(queueName, 0, System.currentTimeMillis(), 0, 1);
            if (zrangeByScore.isEmpty()){
                //如果消息是空的，则暂停500ms
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                //暂停完退出去下一轮
                continue;
                //如果读取到了消息，则直接读取消息
            }

            String next = zrangeByScore.iterator().next();
            if (jedis.zrem(queueName,next)>0){
                //说明获取到了消息，开始处理业务
                try {
                    Message message = new ObjectMapper().readValue(next, Message.class);
                    System.out.println("receive msg:"+message+ new Date());
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }

        }
    }
}
